package atguigu.com.edu.app.dwd.db;

//订单明细

import atguigu.com.edu.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeOrderPreProcess {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(4);

        /*env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        env.getCheckpointConfig().setCheckpointTimeout(5000L);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop201:8020/edu/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15*60 + 10));

        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_order_pre_group"));

        Table detail = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['course_id'] course_id,\n" +
                "    `data`['course_name'] course_name,\n" +
                "    `data`['order_id'] order_id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['origin_amount'] origin_amount,\n" +
                "    `data`['coupon_reduce'] coupon_reduce,\n" +
                "    `data`['final_amount'] final_amount,\n" +
                "     date_format(data['create_time'],'yyyy-MM-dd') date_id,\n" +
                "     data['create_time'] create_time,\n" +
                "ts\n od_ts," +
                "    0 session_id\n" +
                "from topic_db\n" +
                "where `table`='order_detail'\n" +
                "and `type`='insert' ");


        tableEnv.createTemporaryView("detail",detail);
        //   tableEnv.executeSql("select * from detail_info").print();



        Table info = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['origin_amount'] origin_amount,\n" +
                "    `data`['coupon_reduce'] coupon_reduce,\n" +
                "    `data`['final_amount'] final_amount,\n" +
                "    `data`['order_status'] order_status,\n" +
                "    `data`['out_trade_no'] out_trade_no,\n" +
                "    `data`['trade_body'] trade_body,\n" +
                "    `data`['session_id'] session_id,\n" +
                "    `data`['province_id'] province_id,\n" +
                "     date_format(data['create_time'],'yyyy-MM-dd') date_id,\n" +
                "     data['create_time'] create_time,\n" +
                "`type`,\n" +
                "`old`,\n" +
                "     ts oi_ts \n" +
                "from topic_db\n" +
                "where `table`='order_info'\n" +
                "and `type`='insert'or `type` = 'update' ");
        tableEnv.createTemporaryView("info",info);
      //  tableEnv.executeSql("select * from info ").print();

        Table resultTable = tableEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.order_id,\n" +
                "od.origin_amount,\n" +
                "od.coupon_reduce,\n" +
                "od.final_amount,\n" +
                "od.date_id,\n" +
                "od.create_time,\n" +
                "od.od_ts,\n" +


                "oi.user_id,\n" +
                "oi.session_id,\n" +
                "oi.province_id,\n" +
                "date_format(oi.create_time, 'yyyy-MM-dd') operate_date_id,\n" +
                "oi.create_time operate_time,\n" +
                "oi.order_status,\n" +
                "oi.`type`,\n" +
                "oi.`old`,\n" +
                "oi.oi_ts,\n" +

                "current_row_timestamp() row_op_ts\n" +
                "from detail od \n" +
                "join info oi\n" +
                "on od.order_id = oi.id" );
        tableEnv.createTemporaryView("result_table", resultTable);

        tableEnv.executeSql("create table dwd_trade_order_pre_process(\n" +
                "id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "order_id string,\n" +

                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "od_ts string,\n" +


                "user_id string,\n" +
                "session string,\n" +
                "province_id string,\n" +
                "operate_date_id string,\n" +
                "operate_time string,\n" +
                "order_status string,\n" +
                "`type` string,\n" +
                "`old` map<string,string>,\n" +
                "oi_ts string,\n" +


                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));

        tableEnv.executeSql("insert into dwd_trade_order_pre_process select * from result_table");
        tableEnv.executeSql("select * from result_table").print();

    }
}
